package com.example.config;

import com.example.handler.MQTTMessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import javax.annotation.Resource;

/**
 * @Description : MQTT 消费端配置
 * @Author : ZYX
 * @Date : 2023/9/19 16:28
 * @Version : 1.0
 **/
@Configuration
@Slf4j
public class MQTTConsumerConfig {

    @Resource
    private MQTTConfig mqttConfig;

    @Resource
    private MQTTMessageHandler mqttMessageHandler;

    /**
     * 客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息
     */
    private static final byte[] WILL_DATA;

    static {
        WILL_DATA = "offline".getBytes();
    }

    /**
     * 入站直连通道
     * @return
     */
    @Bean("mqttInput")
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * 创建MqttPahoClientFactory 设置MQTT的broker的连接属性
     * @return
     */
    @Bean
    public MqttPahoClientFactory inClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(mqttConfig.getHost());
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        options.setConnectionTimeout(mqttConfig.getTimeout());
        options.setKeepAliveInterval(mqttConfig.getKeepalive());
        options.setCleanSession(mqttConfig.getCleanSession());
        options.setAutomaticReconnect(mqttConfig.getAutomaticReconnect());
        options.setWill("willTopic", WILL_DATA, 2, false);
        factory.setConnectionOptions(options);
        return factory;
    }

    /**
     * 入站
     * @return
     */
    @Bean
    public MessageProducer producer() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId() + "_customer", inClientFactory(), mqttConfig.getTopics());
        adapter.setCompletionTimeout(5000);
        DefaultPahoMessageConverter defaultPahoMessageConverter = new DefaultPahoMessageConverter();
        // 按字节接收消息
        // defaultPahoMessageConverter.setPayloadAsBytes(true);
        adapter.setConverter(defaultPahoMessageConverter);
        adapter.setQos(mqttConfig.getQos());
        adapter.setOutputChannel(mqttInputChannel());

        return adapter;
    }

    /**
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInput")
    public MessageHandler handler() {
        return this.mqttMessageHandler;
    }



}
